package com.systar.activity.report;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.KeeperState;

import com.systar.activity.ZookeeperConnector;

public class ZookeeperMonitor extends ZookeeperConnector
{
	private final ActivityDataPool data;
	private final Set<String> watchedNodes;
	
	public ZookeeperMonitor(String zkHostPort, ActivityDataPool data)
	{
		super(zkHostPort);
		this.data = data;
		watchedNodes = new HashSet<String>();
	}

	@Override
	protected void startupInit() throws Exception
	{
		// Watch znode creation / deletion
		watchNodes();
	}
	
	private void watchNodes() throws InterruptedException, KeeperException
	{
		// Watch znode creation / deletion
		List<String> children = null;
		try
		{
			children = getZooKeeper().getChildren("/", this);
		}
		catch (KeeperException e)
		{
			// KeeperException is thrown if no children, so skip it
		}
		if (children != null)
		{
			for (String path : children)
			{
				if ("zookeeper".equals(path))
					continue;
				if (!watchedNodes.add(path))
					continue;
				updateData(path, getZooKeeper().getData("/"+path, this, null));

			}
		}
	}
	
	@Override
	public void process(WatchedEvent event)
	{
		super.process(event);

		if (event.getState() == KeeperState.Disconnected
				|| event.getState() == KeeperState.Expired
				|| event.getState() == KeeperState.AuthFailed) 
			return;
		
		try
		{
			String path = event.getPath();
			switch (event.getType())
			{
			case NodeDataChanged:
				updateData(path, getZooKeeper().getData(path, this, null));
				break;
			case NodeChildrenChanged:
				if ("/".equals(path))
					watchNodes();
				break;
			default:
				break;
			}
		}
		catch (KeeperException e)
		{
			// TODO error handling
			e.printStackTrace();
		}
		catch (InterruptedException e)
		{
			// TODO error handling
			e.printStackTrace();
		}
	}
	
	private void updateData(String zkPath, byte[] zkData)
	{
		if (zkPath == null || "/".equals(zkPath))
			return;
		
		String source = zkPath.startsWith("/") ? zkPath.substring(1) : zkPath;
		ActivityData activity = new ActivityData(source);
		activity.setData(zkData);
		data.setData(activity); // minimum synchronization overhead
	}

}
